#include "cybertron/network/SocketUdpSender.hpp"
#include "cybertron/network/SocketUdpCommon.hpp"
#include "cybertron/core/UtilCRT.hpp"
#include "cybertron/core/Log.hpp"
#include <thread>
#include <chrono>
#include <queue>
#include <asio.hpp>
#include "cybertron/network/perf.hpp"

CYBERTRON_BEGIN

class SocketUdpSenderImpl
{
public:
	asio::ip::udp::endpoint	mEndPoint;
	std::shared_ptr<asio::ip::udp::socket> mpSocket;
	std::shared_ptr<asio::io_service> mpIoService;
	std::shared_ptr<bool> mThreadRunning;
	std::shared_ptr<std::thread> mpIOServiceThread;
	std::shared_ptr<std::thread> mpSenderThread;
	bool mbSupportMessageSplit;
	std::mutex							mSequenceIdMutex;
	std::uint16_t						mSequenceId;
	std::mutex							mCachedMessageMutex;
	std::queue<std::string>				mCachedMessages;
	std::atomic<bool>					mbMessageSending;
	std::atomic<bool>					mbAsyncMode;
	bool								isMsgEmpty = false;
};

SocketUdpSender::SocketUdpSender(bool bSupportMessageSplit/* = false*/, bool bAsyncMode/* = false*/)
{
	mpImpl = new SocketUdpSenderImpl();
	mpImpl->mbSupportMessageSplit = bSupportMessageSplit;
	mpImpl->mbAsyncMode = bAsyncMode;
}

SocketUdpSender::~SocketUdpSender()
{
	clear();
	delete mpImpl;
	mpImpl = nullptr;
}

bool SocketUdpSender::initialize(const std::string& targetIp, std::uint16_t targetPort, std::uint32_t sendInterval/* = 10us*/)
{
	clear();

	mpImpl->mEndPoint = asio::ip::udp::endpoint(asio::ip::address_v4::from_string(targetIp), targetPort);
	mpImpl->mThreadRunning = std::make_shared<bool>(true);
	mpImpl->mpIoService = std::make_shared<asio::io_service>();
	mpImpl->mpSocket = std::make_shared<asio::ip::udp::socket>(
		*(mpImpl->mpIoService),
		asio::ip::udp::endpoint(asio::ip::udp::v4(), 0));
#if CYBERTRON_WIN || CYBERTRON_LINUX
	asio::socket_base::send_buffer_size option(UDP_MAX_MESSAGE_SIZE);
	mpImpl->mpSocket->set_option(option);
	mpImpl->mpSocket->get_option(option);
	logInfo("Send Buffer Size: %d\r\n", option.value());
#endif
	// start background thread
	auto pThreadRunning = mpImpl->mThreadRunning;
	auto pService = mpImpl->mpIoService;

	if(mpImpl->mbAsyncMode){
		mpImpl->mpIOServiceThread = UtilCRT::startThread(
			"Udp IO Service", 
			[pThreadRunning, pService]()
		{	
			while (*pThreadRunning)
			{
				pService->run();
			}
		});
	}

	mpImpl->mpSenderThread = UtilCRT::startThread(
		"UdpSender",
		[pThreadRunning, this, sendInterval]()
	{
		auto async_send_callback = [this](asio::error_code ec, std::size_t bytes_transferred) {
			if (!ec) {
				IoPerf::inst().incr_send_packs(1);
				IoPerf::inst().incr_send_bytes(bytes_transferred);
			}

			// never resent currently.
			if (mpImpl->mbMessageSending) {
				std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
				mpImpl->mCachedMessages.pop();
				mpImpl->mbMessageSending = false;
			}
		};
		while (*pThreadRunning)
		{
			auto start = std::chrono::high_resolution_clock::now();
			auto end = start + std::chrono::microseconds(sendInterval); // give 10 micro seconds for message send
			if(mpImpl->mbAsyncMode){
				if (mpImpl->mbMessageSending) {
					std::this_thread::sleep_for(std::chrono::milliseconds(1));
					continue;
				}
				{
					std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
					if (mpImpl->mCachedMessages.empty()) 
					{
						mpImpl->isMsgEmpty = true;
					}
					else
					{
						mpImpl->isMsgEmpty = false;
					}
				}
				if (mpImpl->isMsgEmpty)
				{
					std::this_thread::sleep_for(std::chrono::milliseconds(1));
					continue;
				}
				std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
				mpImpl->mbMessageSending = true;
				auto& cachedMessage = mpImpl->mCachedMessages.front();
				//logInfo("Sending Message with size: %d.", cachedMessage.size());
				mpImpl->mpSocket->async_send_to(
					asio::buffer(cachedMessage.c_str(), cachedMessage.size()),
					mpImpl->mEndPoint, async_send_callback);
			}
			else{
				{
					std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
					if (mpImpl->mCachedMessages.empty()) 
					{
						mpImpl->isMsgEmpty = true;
					}
					else
					{
						mpImpl->isMsgEmpty = false;
					}
				}
				if (mpImpl->isMsgEmpty)
				{
					std::this_thread::sleep_for(std::chrono::milliseconds(1));
					continue;
				}
				std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
				auto& cachedMessage = mpImpl->mCachedMessages.front();
				//if(mpImpl->mCachedMessages.size() % 10 == 0){
				//	logInfo("Cached Messages size: %d.", mpImpl->mCachedMessages.size());
				//}
				//logInfo("Sending Message with size: %d.", cachedMessage.size());
				mpImpl->mpSocket->send_to(
					asio::buffer(cachedMessage.c_str(), cachedMessage.size()),
					mpImpl->mEndPoint);

				IoPerf::inst().incr_send_packs(1);
				IoPerf::inst().incr_send_bytes(cachedMessage.size());

				mpImpl->mCachedMessages.pop();
			}
			// yield to lower the packet loss and give receiver some time to consume the data.
			do {
				std::this_thread::yield();
			} while (std::chrono::high_resolution_clock::now() < end);
		}
	});

	return true;
}

void SocketUdpSender::clear()
{
	if (mpImpl->mThreadRunning != nullptr)
	{
		(*mpImpl->mThreadRunning) = false;

		if (mpImpl->mpIoService != nullptr)
		{
			mpImpl->mpIoService->stop();
		}
	}

	if (mpImpl->mpSenderThread != nullptr)
	{
		mpImpl->mpSenderThread->join();
		mpImpl->mpSenderThread = nullptr;
	}
	if (mpImpl->mpIOServiceThread != nullptr)
	{
		mpImpl->mpIOServiceThread->join();
		mpImpl->mpIOServiceThread = nullptr;
	}
	
	mpImpl->mpSocket = nullptr;
	mpImpl->mThreadRunning = nullptr;
	mpImpl->mpIoService = nullptr;
	mpImpl->mSequenceId = 0;
	mpImpl->mbMessageSending = false;
	{
		std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
		mpImpl->mCachedMessages = std::queue<std::string>();
	}
}

void SocketUdpSender::send(const std::string& message)
{
	if (mpImpl->mpSocket == nullptr) {
		return;
	}
	if (mpImpl->mbSupportMessageSplit) {
		std::uint16_t sequenceId;
		{
			std::lock_guard<std::mutex> lock(mpImpl->mSequenceIdMutex);
			++mpImpl->mSequenceId;
			if (mpImpl->mSequenceId == UDP_MAX_FRAGMENT_SIZE) {
				mpImpl->mSequenceId = 0;
			}
			sequenceId = mpImpl->mSequenceId;
			//logDebug("Current sequence Id: %d\r\n\r\n", sequenceId);
		}
		
		std::vector<std::string> encodedMessages = SocketUdpCommon::encodeMessage(message, sequenceId);
		std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
		for (auto& encodedMessage : encodedMessages) {
			mpImpl->mCachedMessages.push(std::move(encodedMessage));
		}
	}
	else if(!message.empty()){
		std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
		mpImpl->mCachedMessages.push(message);
	}
}

void SocketUdpSender::send(const std::vector<std::string>& messages)
{
	if (mpImpl->mpSocket == nullptr) {
		return;
	}
	if (mpImpl->mbSupportMessageSplit) {
		for (const auto& message : messages) {
			send(message);
		}
	}
	else {
		std::lock_guard<std::mutex> lock(mpImpl->mCachedMessageMutex);
		for (auto& encodedMessage : messages) {
			mpImpl->mCachedMessages.push(std::move(encodedMessage));
		}
	}
}

void SocketUdpSender::send(const Message& msg)
{
	std::string stringMessage((const char*)msg.getData(), msg.getLength());
	send(stringMessage);
}

CYBERTRON_END
